/**
 * @file StreamReader.c
 *
 * This file implements the StreamReader interface.
 *
 * @author Luc Fourestier
 * @date 2009
 *
 * @page SR The StreamReader
 *
 * The StreamReader is a "catcher" of streamed data that can understand heterogeneous input type.
 * The StreamReader will take the input given by the options, open/start the stream from that input,
 * and push it to a StreamAnalyzer.
 *
 * @section SR_Arch The Architecture
 *
 * The StreamReader is based on a thread that will open/start the source (UDP, TCP, file...)
 * and read, chunk after chunk, the stream.
 * The chunks are then pushed into the circular buffer towards the StreamAnanalyzer.
 * No treatment is done on the stream there except inserting a tag. This tagging
 * is used to pass information on the related chunk.
 *
 * @section SR_Design The design
 *
 * The StreamReader is created via SR_Create() using SR_Option_s as input.
 * This function will allocate and return a SR_StreamReader_s structure that
 * contains all the elements required by the StreamReader to run. This structure
 * is used to instantiate the StreamReader as many time as we want.
 *
 * The StreamReader MUST be linked to a StreamAnanalyzer. Is is done thanks to
 * the "CB" field of the SR_Option_s structure. This field must be filled
 * with the CircBuff_s pointer of the StreamAnalyzer you want to link with.
 *
 * Once this link is done the StreamReader is ready to work.
 *
 * The options specify which type of input the StreamReader will read (see @ref InOutType_e).
 * According to the input, the StreamReader will start one of the following thread.
 *
 * @subsection SR_Design_File The FileThread
 *
 * In the case of file reading, the SR_Create() will spawn the FileThread() thread.
 * It starts directly to read the file(s) and to push the chunks into the circular buffer.
 * Some end of file detection has been implemented. Indeed, the end of file (and so stream)
 * must be passed to the streamAnalyzer. In such case, the FileThread() will insert a null
 * packet at the end specifying that this chunk is last in the tag (see Tag_s). This will stop
 * the stream in the triplette.
 * In the case of playlist, the end of the stream will be signalled only at the end of the last
 * file.
 *
 * @subsection SR_Design_UDP The UDPServerThread
 *
 * In the case of UDP input, the SR_Create() will spawn the UDPServerThread() thread.
 * This thread open the proper socket (also muticast if the address is in the multicast
 * range) and wait for UDP packet reception. When receiving packets, it will tag them with
 * the reception timestamp and push them into the StreamAnanalyzer.
 *
 * @subsection SR_Design_TCP The TCPServerThread
 *
 * In the case of TCP input, the SR_Create() will spawn the TCPServerThread() thread.
 * This thread open a listen socket and wait for TCP packet reception. When receiving
 * packets, it will tag them with the reception timestamp and push them into the
 * StreamAnanalyzer.
 */

#include "StreamReader.h"

#include "TS.h"

#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <sys/socket.h>

// Network connection default config
#define MAX_UDP_BUFF_SIZE MAX_CHUNK_SIZE
#define MAX_TCP_BUFF_SIZE MAX_CHUNK_SIZE
#define MAX_RTP_BUFF_SIZE MAX_CHUNK_SIZE
#define MC_LOW_ADD        0xe00000ff
#define	MC_HIGH_ADD       0xefffffff
#define SOCKET_RCVBUFF_SIZE 150000


// Statics
static void * UDPServerThread(void * Arg);
static void * TCPServerThread(void * Arg);
static void * FileThread(void * Arg);

/**
 * Create the StreamReader.
 *
 * @param SR_p (out): The pointer where to return the creates StreamReader
 * @param Options (in): The options of creation
 * @return Errors: SR_ERROR_NOT_SUPPORTED if input type is not supported yet!
 */
int SR_Create(SR_StreamReader_t ** SR_p, SR_Option_t Options)
{
	int RetVal = OK;
	SR_StreamReader_t * SR = NULL;
	unsigned int i = 0;

	if ((SR_p != NULL) && (Options.CB != NULL))
	{
		// Create the SR
		SR = (SR_StreamReader_t*) malloc(sizeof(SR_StreamReader_t));
		if (SR == NULL)
		{
			return(SR_ERROR_CANNOT_ALLOCATE);
		}
		bzero((void*)SR, sizeof(SR_StreamReader_t));

		// Copy the options
		if ((Options.BindAddress[0] == 0) || (strlen(Options.BindAddress) < 7))
		{
			strncpy(Options.BindAddress, "0.0.0.0", IP_ADDR_LENGTH);
		}
		SR->Options = Options;

		// Start the thread
		SR->ThreadLoop = 1;
		switch (SR->Options.InputType)
		{
			case File:
				i = 0;
				while ((SR->Options.FileName[i][0] != 0) && (i < SR_MAX_FILE))
				{
					SR->InputFile[i] = fopen(SR->Options.FileName[i], "r");
					if (SR->InputFile[i] == NULL)
					{
						break;
					}

					i++;
				}

				if (i == 0)
				{
					return(SR_ERROR_FILE_PB);
				}
				SR->FileNumber = i;

				if (pthread_create(&SR->ThreadID, NULL, FileThread, (void*) SR) != 0)
				{
					return(SR_ERROR_CANNOT_ALLOCATE);
				}
				break;

			case Tcp:
				if (pthread_create(&SR->ThreadID, NULL, TCPServerThread, (void*) SR) != 0)
				{
					return(SR_ERROR_CANNOT_ALLOCATE);
				}
				break;

			case Udp:
				if (pthread_create(&SR->ThreadID, NULL, UDPServerThread, (void*) SR) != 0)
				{
					return(SR_ERROR_CANNOT_ALLOCATE);
				}
				break;

			default:
				return(SR_ERROR_NOT_SUPPORTED);
				break;
		}

		// Return the structure
		*SR_p = SR;
	}
	else
	{
		RetVal = SR_ERROR_BAD_PARAM;
	}

	return(RetVal);
}


/**
 * Delete the StreamReader.
 *
 * @param SR (in): The StreamReader
 * @return Errors
 */
int SR_Delete(SR_StreamReader_t * SR)
{
	int RetVal = OK;
	unsigned int i = 0;

	if (SR != NULL)
	{
		// Kill thread
		SR->ThreadLoop = 0;
		CircB_KillWrite(SR->Options.CB);

		switch (SR->Options.InputType)
		{
			case File:
				// close the input file
				i = 0;
				while (i < SR->FileNumber)
				{
					if (SR->InputFile[i] != NULL)
					{
						fclose(SR->InputFile[i]);
						SR->InputFile[i] = NULL;
					}
					i++;
				}
				break;

			case Tcp:
				shutdown(SR->Socket, SHUT_RDWR);
				shutdown(SR->ListenSocket, SHUT_RDWR);
				break;

			case Udp:
				shutdown(SR->Socket, SHUT_RDWR);
				break;

			default:
				break;
		}
		pthread_join(SR->ThreadID, NULL);

		// Free StreamReader
		free(SR);
	}
	else
	{
		RetVal = SR_ERROR_BAD_PARAM;
	}

	return(RetVal);
}

/**
 * Print the StreamReader info structure
 *
 * @param SR (in): The SR to print
 * @param Name (in): The name of the SR (any nickname you wish!).
 * @param level (in): Level of details (Lowest = 0).
 * @param Output (in): Where to print (stdout or FILE).
 * @return Errors
 */
int SR_Print(SR_StreamReader_t * SR, char * Name, int Level, FILE * Output)
{
	int RetVal = OK;
	unsigned int i = 0;

	if ((SR == NULL) && (Output == NULL))
	{
		return(ERROR);
	}

	if (Name == NULL)
	{
		Name = "StreamReader";
	}

	switch (SR->Options.InputType)
	{
	case File:
		fprintf(Output, "%s input file(s): ", Name);
		i = 0;
		while (i < SR->FileNumber)
		{
			if (i == SR->CurrentFile)
			{
				fprintf(Output, "*");
			}
			fprintf(Output, "%s ", SR->Options.FileName[i]);
			i++;
		}
		fprintf(Output, "\n");

		break;

	case Udp:
		fprintf(Output, "%s input address: %s:%i\n", Name, SR->Options.RcvAddress, SR->Options.RcvPort);
		fprintf(Output, "%s own address: %s\n", Name, SR->Options.BindAddress);
		break;

	default:
		break;
	}

	fprintf(Output, "%s total bytes read: %i\n", Name, (int)SR->TotalByteRead);

	// Internals status
	if (Level >= PRINT_LEVEL_INTERNALS)
	{
		fprintf(Output, "%s thread state: %i\n", Name, (int)SR->ThreadLoop);
		if (SR->Options.InputType == Udp)
		{
			fprintf(Output, "%s socket buffer size: %i\n", Name, (int)SR->RcvBuffSize);

		}

#ifdef DEBUG_SR_PROFILING
		if (SR->passCount > 1)
		{
			fprintf(Output, "%s total pass duration: %is %iµs\n", Name, (int)SR->TotalPassDuration.tv_sec, (int)SR->TotalPassDuration.tv_usec);
			fprintf(Output, "%s pass count: %lli\n", Name, SR->passCount-1);
			fprintf(Output, "%s Average duration per pass: %lli µs\n", Name, TIMEVAL_TO_USECONDS(&SR->TotalPassDuration)/(SR->passCount-1));
			fprintf(Output, "%s pass errors: %lli\n", Name, SR->passError);
		}
#endif
	}

	fprintf(Output, "\n");

	return(RetVal);
}

/**
 * The UDP socket reception thread
 *
 * @param Arg: the thread argument
 */
static void * UDPServerThread(void * Arg)
{
	int Error = 0;
	SR_StreamReader_t * SR = (SR_StreamReader_t*) Arg;
	void * Buff_p = NULL;
    struct ip_mreq mreq;
    unsigned int opt = 1;
    unsigned int OptSize = sizeof(opt);

	if (SR == NULL)
	{
		pthread_exit(NULL);
		return(NULL);
	}

    // Create socket
	SR->Socket = socket(AF_INET, SOCK_DGRAM, 0);
    if (SR->Socket < 0)
    {
    	pthread_exit(NULL);
    }

	// Set reusability
    if (setsockopt(SR->Socket, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) == -1)
	{
	   	pthread_exit(NULL);
	}

    // Set buffer size
    opt = SOCKET_RCVBUFF_SIZE;
	if (setsockopt(SR->Socket, SOL_SOCKET, SO_RCVBUF, (void*) &opt, OptSize) == -1)
	{
	   	pthread_exit(NULL);
	}

    if (getsockopt(SR->Socket, SOL_SOCKET, SO_RCVBUF, (void*) &opt, &OptSize) == -1)
	{
	   	pthread_exit(NULL);
	}
    SR->RcvBuffSize = opt;

    // Bind
    bzero((char *) &SR->LocalAddr, sizeof(SR->LocalAddr));
    SR->LocalAddr.sin_family = AF_INET;
    SR->LocalAddr.sin_addr.s_addr = inet_addr(SR->Options.BindAddress);
    SR->LocalAddr.sin_port = htons(SR->Options.RcvPort);
    if (bind(SR->Socket, (struct sockaddr *) &SR->LocalAddr, sizeof(SR->LocalAddr)) < 0)
    {
    	pthread_exit(NULL);
    }

	// Join multicast
    unsigned int MCAddr = ntohl(inet_addr(SR->Options.RcvAddress));
    if ((MCAddr >= MC_LOW_ADD) && (MCAddr <= MC_HIGH_ADD))
    {
        mreq.imr_multiaddr.s_addr = inet_addr(SR->Options.RcvAddress);
        mreq.imr_interface.s_addr = inet_addr(SR->Options.BindAddress);
    	setsockopt(SR->Socket, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq, sizeof(mreq));
    }

#ifdef DEBUG_SR_PROFILING
	struct timeval Start;
	Start.tv_sec = 0;
	Start.tv_usec = 0;
#endif

    // Thread loop
    while(SR->ThreadLoop)
	{
		int Len = 0;
	    int RecvCount = 1;
	    unsigned int BuffSize = 0;

#ifdef DEBUG_SR_PROFILING
		struct timeval New;
		gettimeofday(&New, NULL);
		if (Start.tv_sec == 0)
		{
			Start = New;
		}
		timersub(&New, &Start, &SR->TotalPassDuration);
		SR->passCount++;
#endif

		// Receive chunks
		Error = CircB_GetWriteElt(SR->Options.CB, &Buff_p, &BuffSize, CIRCB_WAIT_FOREVER);
		if (Error == OK)
		{
			bzero(Buff_p, BuffSize);

			Len = sizeof(struct sockaddr_in);
			RecvCount = recvfrom(SR->Socket, (void *)((unsigned char*)Buff_p+sizeof(Tag_t)), BuffSize-sizeof(Tag_t), 0, (struct sockaddr*) &SR->ClientAddr, (socklen_t *) &Len);
			if (RecvCount > 0)
			{
				unsigned int Size = RecvCount+sizeof(Tag_t);
				struct timeval NewTime;

				// Time stamp
				gettimeofday(&NewTime, NULL);
				timersub(&NewTime, &SR->Options.ReferenceTime, &NewTime); // Relative time
				((Tag_t*)Buff_p)->ReceptionTime = NewTime;

				// Count
				SR->TotalByteRead += RecvCount;

				Error = CircB_ValidateWriteElt(SR->Options.CB, Size);
			}
			else
			{
				Error = CircB_ReleaseWriteElt(SR->Options.CB);
			}
		}

#ifdef DEBUG_SR_PROFILING
		if (Error != OK)
		{
			SR->passError++;
		}
		Error = OK;
#endif
	}

    pthread_exit(NULL);
}

/**
 * The TCP socket reception thread
 *
 * @param Arg: the thread argument
 */
static void * TCPServerThread(void * Arg)
{
	int Error = 0;
	SR_StreamReader_t * SR = (SR_StreamReader_t*) Arg;
	void * Buff_p = NULL;
    unsigned int opt = 1;
    unsigned int OptSize = sizeof(opt);

    // Create main socket
	SR->ListenSocket = socket(AF_INET, SOCK_STREAM, 0);
    if (SR->ListenSocket < 0)
    {
    	pthread_exit(NULL);
    }

	if (setsockopt(SR->ListenSocket, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) == -1)
	{
	   	pthread_exit(NULL);
	}

    // Bind
    bzero((char *) &SR->LocalAddr, sizeof(SR->LocalAddr));
    SR->LocalAddr.sin_family = AF_INET;
    SR->LocalAddr.sin_addr.s_addr = inet_addr(SR->Options.BindAddress);
    SR->LocalAddr.sin_port = htons(SR->Options.RcvPort);
    if (bind(SR->ListenSocket, (struct sockaddr *) &SR->LocalAddr, sizeof(SR->LocalAddr)) < 0)
    {
    	pthread_exit(NULL);
    }

    // Listen
    if(listen(SR->ListenSocket, 1) < 0)
    {
    	pthread_exit(NULL);
    }

#ifdef DEBUG_SR_PROFILING
	struct timeval Start;
	Start.tv_sec = 0;
	Start.tv_usec = 0;
#endif

    while(SR->ThreadLoop)
	{
		int Len = 0;
	    int RecvCount = 1;
	    unsigned int BuffSize = 0;

		// Accept
	    Len = sizeof(SR->ClientAddr);
	    SR->Socket = accept(SR->ListenSocket, (struct sockaddr *) &SR->ClientAddr, (socklen_t*)&Len);
		if (SR->Socket < 0)
		{
			pthread_exit(NULL);
		}

	    // Set buffer size
	    opt = SOCKET_RCVBUFF_SIZE;
		if (setsockopt(SR->Socket, SOL_SOCKET, SO_RCVBUF, (void*) &opt, OptSize) == -1)
		{
		   	pthread_exit(NULL);
		}

		// Process communication
		while (RecvCount > 0)
		{
#ifdef DEBUG_SR_PROFILING
			struct timeval New;
			gettimeofday(&New, NULL);
			if (Start.tv_sec == 0)
			{
				Start = New;
			}
			timersub(&New, &Start, &SR->TotalPassDuration);
			SR->passCount++;
#endif

			Error = CircB_GetWriteElt(SR->Options.CB, &Buff_p, &BuffSize, CIRCB_WAIT_FOREVER);
			if (Error == OK)
			{
				bzero(Buff_p, BuffSize);
				RecvCount = recv(SR->Socket, (void *)((unsigned char*)Buff_p+sizeof(Tag_t)), BuffSize-sizeof(Tag_t), 0);
				if (RecvCount > 0)
				{
					unsigned int Size = RecvCount+sizeof(Tag_t);
					struct timeval NewTime;

					// Time stamp
					gettimeofday(&NewTime, NULL);
					timersub(&NewTime, &SR->Options.ReferenceTime, &NewTime); // Relative time
					((Tag_t*)Buff_p)->ReceptionTime = NewTime;

					// Count
					SR->TotalByteRead += RecvCount;

					Error = CircB_ValidateWriteElt(SR->Options.CB, Size);
				}
				else
				{
					Error = CircB_ReleaseWriteElt(SR->Options.CB);
				}
			}

#ifdef DEBUG_SR_PROFILING
			if (Error != OK)
			{
				SR->passError++;
			}
			Error = OK;
#endif
		}

		shutdown(SR->Socket, SHUT_RDWR);
	}

	pthread_exit(NULL);
}


/**
 * The file reading thread
 *
 * @param Arg: the thread argument
 */
static void * FileThread(void * Arg)
{
	int Error = 0;
	SR_StreamReader_t * SR = (SR_StreamReader_t*) Arg;
    void * Buff_p = NULL;
    unsigned int BuffSize = 0;
	unsigned int ReadCount = 0;

    if (SR->InputFile[0] == NULL)
    {
    	pthread_exit(NULL);
    	return(NULL);
    }

#ifdef DEBUG_SR_PROFILING
	struct timeval Start;
	Start.tv_sec = 0;
	Start.tv_usec = 0;
#endif

    while (SR->ThreadLoop)
	{
		unsigned int Size = 0;

#ifdef DEBUG_SR_PROFILING
		struct timeval New;
		gettimeofday(&New, NULL);
		if (Start.tv_sec == 0)
		{
			Start = New;
		}
		timersub(&New, &Start, &SR->TotalPassDuration);
		SR->passCount++;
#endif

		Error = CircB_GetWriteElt(SR->Options.CB, &Buff_p, &BuffSize, CIRCB_WAIT_FOREVER);
		if (Error == OK)
		{
			bzero(Buff_p, BuffSize);
			if (SR->InputFile[SR->CurrentFile] != NULL)
			{
				ReadCount = fread((void*)((unsigned char*)Buff_p+sizeof(Tag_t)), 1, BuffSize-sizeof(Tag_t), SR->InputFile[SR->CurrentFile]);
			}
			else
			{
				ReadCount = 0;
				SR->CurrentFile = SR->FileNumber; // A way to kill the process properly
			}

			// insert a Null packet if nothing has been read to prevent problems in the stream.
			if (ReadCount == 0)
			{
				bcopy((void*)NullPacket, (void*)((unsigned char*)Buff_p+sizeof(Tag_t)), TS_SIZE);
				Size = TS_SIZE + sizeof(Tag_t);
			}
			else
			{
				Size = ReadCount + sizeof(Tag_t);
				SR->TotalByteRead += ReadCount;
			}

			if ((ReadCount == 0) || (feof(SR->InputFile[SR->CurrentFile]) || (ferror(SR->InputFile[SR->CurrentFile])))) // End of file
			{
				// Next file in playlist
				SR->CurrentFile++;
				((Tag_t*)Buff_p)->ChunkIsLast = 1;

				if (SR->CurrentFile >= SR->FileNumber) // End of stream
				{
					SR->ThreadLoop = 0;
				}
			}

			Error = CircB_ValidateWriteElt(SR->Options.CB, Size);
		}

#ifdef DEBUG_SR_PROFILING
		if (Error != OK)
		{
			SR->passError++;
		}
		Error = OK;
#endif
	}

	pthread_exit(NULL);
}


